//! 哈希槽管理模块
//! 
//! 实现Redis集群的16384个哈希槽分片机制

use crate::cluster::traits::{SlotManager as SlotManagerTrait, SlotStats};
use crate::cluster::{ClusterError, ClusterResult};
use async_trait::async_trait;
use bit_vec::BitVec;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

/// Redis集群总槽位数
pub const CLUSTER_SLOTS: u16 = 16384;

/// 槽位迁移信息
#[derive(Debug, Clone)]
pub struct MigrationInfo {
    /// 源节点ID
    pub source_node: String,
    /// 目标节点ID  
    pub target_node: String,
    /// 迁移状态
    pub state: MigrationState,
    /// 开始时间
    pub started_at: std::time::Instant,
    /// 已迁移键数量
    pub migrated_keys: usize,
    /// 总键数量
    pub total_keys: usize,
}

/// 迁移状态
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum MigrationState {
    /// 准备中
    Preparing,
    /// 迁移中
    Migrating,
    /// 完成
    Completed,
    /// 失败
    Failed,
    /// 已取消
    Cancelled,
}

/// 槽位状态
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SlotState {
    /// 正常状态 - 槽位被某个节点拥有
    Stable(String),
    /// 迁移中 - 正在从源节点迁移到目标节点
    Migrating { from: String, to: String },
    /// 导入中 - 正在接收来自其他节点的槽位
    Importing { from: String, to: String },
    /// 未分配
    Unassigned,
}

/// 哈希槽管理器
#[derive(Debug)]
pub struct SlotManager {
    /// 槽位到节点的映射 (每个槽位对应一个节点ID)
    slot_to_node: [Option<String>; CLUSTER_SLOTS as usize],
    
    /// 节点到槽位的映射 (每个节点拥有的槽位集合)
    node_to_slots: HashMap<String, BitVec>,
    
    /// 迁移中的槽位信息
    migrating_slots: HashMap<u16, MigrationInfo>,
    
    /// 导入中的槽位信息  
    importing_slots: HashMap<u16, MigrationInfo>,
    
    /// 槽位状态缓存
    slot_states: HashMap<u16, SlotState>,
}

impl Default for SlotManager {
    fn default() -> Self {
        Self::new()
    }
}

impl SlotManager {
    /// 创建新的槽位管理器
    pub fn new() -> Self {
        Self {
            slot_to_node: std::array::from_fn(|_| None),
            node_to_slots: HashMap::new(),
            migrating_slots: HashMap::new(),
            importing_slots: HashMap::new(),
            slot_states: HashMap::new(),
        }
    }
    
    /// 使用CRC16算法计算键的槽位
    pub fn calculate_slot(key: &str) -> u16 {
        // 检查是否有hash tag (大括号)
        let hash_key = if let Some(start) = key.find('{') {
            if let Some(end) = key[start + 1..].find('}') {
                let tag = &key[start + 1..start + 1 + end];
                if !tag.is_empty() {
                    tag
                } else {
                    key
                }
            } else {
                key
            }
        } else {
            key
        };
        
        Self::crc16(hash_key.as_bytes()) % CLUSTER_SLOTS
    }
    
    /// CRC16算法实现 (Redis使用的CRC16-ANSI版本)
    fn crc16(data: &[u8]) -> u16 {
        const CRC16_TAB: [u16; 256] = [
            0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50a5, 0x60c6, 0x70e7,
            0x8108, 0x9129, 0xa14a, 0xb16b, 0xc18c, 0xd1ad, 0xe1ce, 0xf1ef,
            0x1231, 0x0210, 0x3273, 0x2252, 0x52b5, 0x4294, 0x72f7, 0x62d6,
            0x9339, 0x8318, 0xb37b, 0xa35a, 0xd3bd, 0xc39c, 0xf3ff, 0xe3de,
            0x2462, 0x3443, 0x0420, 0x1401, 0x64e6, 0x74c7, 0x44a4, 0x5485,
            0xa56a, 0xb54b, 0x8528, 0x9509, 0xe5ee, 0xf5cf, 0xc5ac, 0xd58d,
            0x3653, 0x2672, 0x1611, 0x0630, 0x76d7, 0x66f6, 0x5695, 0x46b4,
            0xb75b, 0xa77a, 0x9719, 0x8738, 0xf7df, 0xe7fe, 0xd79d, 0xc7bc,
            0x48c4, 0x58e5, 0x6886, 0x78a7, 0x0840, 0x1861, 0x2802, 0x3823,
            0xc9cc, 0xd9ed, 0xe98e, 0xf9af, 0x8948, 0x9969, 0xa90a, 0xb92b,
            0x5af5, 0x4ad4, 0x7ab7, 0x6a96, 0x1a71, 0x0a50, 0x3a33, 0x2a12,
            0xdbfd, 0xcbdc, 0xfbbf, 0xeb9e, 0x9b79, 0x8b58, 0xbb3b, 0xab1a,
            0x6ca6, 0x7c87, 0x4ce4, 0x5cc5, 0x2c22, 0x3c03, 0x0c60, 0x1c41,
            0xedae, 0xfd8f, 0xcdec, 0xddcd, 0xad2a, 0xbd0b, 0x8d68, 0x9d49,
            0x7e97, 0x6eb6, 0x5ed5, 0x4ef4, 0x3e13, 0x2e32, 0x1e51, 0x0e70,
            0xff9f, 0xefbe, 0xdfdd, 0xcffc, 0xbf1b, 0xaf3a, 0x9f59, 0x8f78,
            0x9188, 0x81a9, 0xb1ca, 0xa1eb, 0xd10c, 0xc12d, 0xf14e, 0xe16f,
            0x1080, 0x00a1, 0x30c2, 0x20e3, 0x5004, 0x4025, 0x7046, 0x6067,
            0x83b9, 0x9398, 0xa3fb, 0xb3da, 0xc33d, 0xd31c, 0xe37f, 0xf35e,
            0x02b1, 0x1290, 0x22f3, 0x32d2, 0x4235, 0x5214, 0x6277, 0x7256,
            0xb5ea, 0xa5cb, 0x95a8, 0x8589, 0xf56e, 0xe54f, 0xd52c, 0xc50d,
            0x34e2, 0x24c3, 0x14a0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405,
            0xa7db, 0xb7fa, 0x8799, 0x97b8, 0xe75f, 0xf77e, 0xc71d, 0xd73c,
            0x26d3, 0x36f2, 0x0691, 0x16b0, 0x6657, 0x7676, 0x4615, 0x5634,
            0xd94c, 0xc96d, 0xf90e, 0xe92f, 0x99c8, 0x89e9, 0xb98a, 0xa9ab,
            0x5844, 0x4865, 0x7806, 0x6827, 0x18c0, 0x08e1, 0x3882, 0x28a3,
            0xcb7d, 0xdb5c, 0xeb3f, 0xfb1e, 0x8bf9, 0x9bd8, 0xabbb, 0xbb9a,
            0x4a75, 0x5a54, 0x6a37, 0x7a16, 0x0af1, 0x1ad0, 0x2ab3, 0x3a92,
            0xfd2e, 0xed0f, 0xdd6c, 0xcd4d, 0xbdaa, 0xad8b, 0x9de8, 0x8dc9,
            0x7c26, 0x6c07, 0x5c64, 0x4c45, 0x3ca2, 0x2c83, 0x1ce0, 0x0cc1,
            0xef1f, 0xff3e, 0xcf5d, 0xdf7c, 0xaf9b, 0xbfba, 0x8fd9, 0x9ff8,
            0x6e17, 0x7e36, 0x4e55, 0x5e74, 0x2e93, 0x3eb2, 0x0ed1, 0x1ef0
        ];
        
        let mut crc = 0u16;
        for &byte in data {
            crc = (crc << 8) ^ CRC16_TAB[((crc >> 8) ^ byte as u16) as usize & 0xff];
        }
        crc
    }
    
    /// 获取槽位对应的节点
    pub fn get_node_for_slot(&self, slot: u16) -> Option<&String> {
        if slot < CLUSTER_SLOTS {
            self.slot_to_node[slot as usize].as_ref()
        } else {
            None
        }
    }
    
    /// 分配槽位给节点
    pub fn assign_slot(&mut self, slot: u16, node_id: String) -> Result<(), String> {
        if slot >= CLUSTER_SLOTS {
            return Err(format!("无效的槽位: {slot}"));
        }
        
        // 检查槽位是否已被分配
        if let Some(existing_node) = &self.slot_to_node[slot as usize] {
            if existing_node != &node_id {
                return Err(format!("槽位 {slot} 已被节点 {existing_node} 拥有"));
            }
        }
        
        // 分配槽位
        self.slot_to_node[slot as usize] = Some(node_id.clone());
        
        // 更新节点槽位映射
        let node_slots = self.node_to_slots.entry(node_id.clone()).or_insert_with(|| {
            BitVec::from_elem(CLUSTER_SLOTS as usize, false)
        });
        node_slots.set(slot as usize, true);
        
        // 更新槽位状态
        self.slot_states.insert(slot, SlotState::Stable(node_id));
        
        Ok(())
    }
    
    /// 取消分配槽位
    pub fn unassign_slot(&mut self, slot: u16) -> Result<Option<String>, String> {
        if slot >= CLUSTER_SLOTS {
            return Err(format!("无效的槽位: {slot}"));
        }
        
        let removed_node = self.slot_to_node[slot as usize].take();
        
        if let Some(node_id) = &removed_node {
            // 更新节点槽位映射
            if let Some(node_slots) = self.node_to_slots.get_mut(node_id) {
                node_slots.set(slot as usize, false);
            }
        }
        
        // 更新槽位状态
        self.slot_states.insert(slot, SlotState::Unassigned);
        
        Ok(removed_node)
    }
    
    /// 获取节点拥有的槽位
    pub fn get_slots_for_node(&self, node_id: &str) -> Vec<u16> {
        if let Some(slots) = self.node_to_slots.get(node_id) {
            slots
                .iter()
                .enumerate()
                .filter_map(|(i, bit)| if bit { Some(i as u16) } else { None })
                .collect()
        } else {
            Vec::new()
        }
    }
    
    /// 获取节点拥有的槽位数量
    pub fn get_slot_count_for_node(&self, node_id: &str) -> usize {
        if let Some(slots) = self.node_to_slots.get(node_id) {
            slots.iter().filter(|&bit| bit).count()
        } else {
            0
        }
    }
    
    /// 获取所有已分配的槽位
    pub fn get_assigned_slots(&self) -> Vec<u16> {
        (0..CLUSTER_SLOTS)
            .filter(|&slot| self.slot_to_node[slot as usize].is_some())
            .collect()
    }
    
    /// 获取所有未分配的槽位
    pub fn get_unassigned_slots(&self) -> Vec<u16> {
        (0..CLUSTER_SLOTS)
            .filter(|&slot| self.slot_to_node[slot as usize].is_none())
            .collect()
    }
    
    /// 检查所有槽位是否已分配
    pub fn all_slots_assigned(&self) -> bool {
        self.get_unassigned_slots().is_empty()
    }
    
    /// 开始槽位迁移
    pub fn start_migration(&mut self, slot: u16, from_node: String, to_node: String) -> Result<(), String> {
        if slot >= CLUSTER_SLOTS {
            return Err(format!("无效的槽位: {slot}"));
        }
        
        // 检查槽位当前归属
        if let Some(current_node) = &self.slot_to_node[slot as usize] {
            if current_node != &from_node {
                return Err(format!("槽位 {slot} 不属于节点 {from_node}"));
            }
        } else {
            return Err(format!("槽位 {slot} 未分配"));
        }
        
        // 检查是否已在迁移中
        if self.migrating_slots.contains_key(&slot) || self.importing_slots.contains_key(&slot) {
            return Err(format!("槽位 {slot} 已在迁移中"));
        }
        
        let migration_info = MigrationInfo {
            source_node: from_node.clone(),
            target_node: to_node.clone(),
            state: MigrationState::Preparing,
            started_at: std::time::Instant::now(),
            migrated_keys: 0,
            total_keys: 0,
        };
        
        self.migrating_slots.insert(slot, migration_info.clone());
        self.importing_slots.insert(slot, migration_info);
        
        // 更新槽位状态
        self.slot_states.insert(slot, SlotState::Migrating { 
            from: from_node, 
            to: to_node 
        });
        
        Ok(())
    }
    
    /// 完成槽位迁移
    pub fn complete_migration(&mut self, slot: u16) -> Result<(), String> {
        if slot >= CLUSTER_SLOTS {
            return Err(format!("无效的槽位: {slot}"));
        }
        
        let migration_info = self.migrating_slots.remove(&slot)
            .ok_or_else(|| format!("槽位 {slot} 不在迁移中"))?;
        
        self.importing_slots.remove(&slot);
        
        // 将槽位分配给目标节点
        self.unassign_slot(slot)?;
        self.assign_slot(slot, migration_info.target_node)?;
        
        Ok(())
    }
    
    /// 取消槽位迁移
    pub fn cancel_migration(&mut self, slot: u16) -> Result<(), String> {
        if slot >= CLUSTER_SLOTS {
            return Err(format!("无效的槽位: {slot}"));
        }
        
        let migration_info = self.migrating_slots.remove(&slot)
            .ok_or_else(|| format!("槽位 {slot} 不在迁移中"))?;
        
        self.importing_slots.remove(&slot);
        
        // 恢复槽位状态
        self.slot_states.insert(slot, SlotState::Stable(migration_info.source_node));
        
        Ok(())
    }
    
    /// 获取槽位状态
    pub fn get_slot_state(&self, slot: u16) -> SlotState {
        if slot >= CLUSTER_SLOTS {
            return SlotState::Unassigned;
        }
        
        self.slot_states.get(&slot).cloned().unwrap_or_else(|| {
            if let Some(node_id) = &self.slot_to_node[slot as usize] {
                SlotState::Stable(node_id.clone())
            } else {
                SlotState::Unassigned
            }
        })
    }
    
    /// 批量分配槽位给节点 (用于集群初始化)
    pub fn batch_assign_slots(&mut self, node_slots: HashMap<String, Vec<u16>>) -> Result<(), String> {
        for (node_id, slots) in node_slots {
            for slot in slots {
                self.assign_slot(slot, node_id.clone())?;
            }
        }
        Ok(())
    }
    
    /// 自动分配槽位 (平均分配给指定节点)
    pub fn auto_assign_slots(&mut self, node_ids: &[String]) -> Result<(), String> {
        if node_ids.is_empty() {
            return Err("节点列表不能为空".to_string());
        }
        
        let slots_per_node = CLUSTER_SLOTS as usize / node_ids.len();
        let remainder = CLUSTER_SLOTS as usize % node_ids.len();
        
        let mut current_slot = 0u16;
        
        for (i, node_id) in node_ids.iter().enumerate() {
            let mut slot_count = slots_per_node;
            if i < remainder {
                slot_count += 1;
            }
            
            for _ in 0..slot_count {
                self.assign_slot(current_slot, node_id.clone())?;
                current_slot += 1;
            }
        }
        
        Ok(())
    }
    
    /// 获取已分配的槽位数量
    pub fn assigned_slot_count(&self) -> usize {
        self.slot_to_node.iter().filter(|slot| slot.is_some()).count()
    }
    
    /// 获取迁移中的槽位列表
    pub fn get_migrating_slots(&self) -> Vec<u16> {
        self.migrating_slots.keys().copied().collect()
    }
    
    /// 获取槽位分布统计
    pub fn get_slot_distribution(&self) -> HashMap<String, usize> {
        let mut distribution = HashMap::new();
        
        for node_id in self.node_to_slots.keys() {
            let count = self.get_slot_count_for_node(node_id);
            distribution.insert(node_id.clone(), count);
        }
        
        distribution
    }
    
    /// 生成槽位信息字符串 (类似Redis CLUSTER SLOTS格式)
    pub fn to_slots_info(&self) -> Vec<(u16, u16, String)> {
        let mut ranges = Vec::new();
        let mut current_start = None;
        let mut current_end = None;
        let mut current_node = None;
        
        for slot in 0..CLUSTER_SLOTS {
            if let Some(node_id) = &self.slot_to_node[slot as usize] {
                if current_node.as_ref() == Some(node_id) && current_end == Some(slot - 1) {
                    // 继续当前范围
                    current_end = Some(slot);
                } else {
                    // 保存上一个范围
                    if let (Some(start), Some(end), Some(node)) = (current_start, current_end, current_node) {
                        ranges.push((start, end, node));
                    }
                    // 开始新范围
                    current_start = Some(slot);
                    current_end = Some(slot);
                    current_node = Some(node_id.clone());
                }
            } else {
                // 保存上一个范围
                if let (Some(start), Some(end), Some(node)) = (current_start, current_end, current_node) {
                    ranges.push((start, end, node));
                }
                current_start = None;
                current_end = None;
                current_node = None;
            }
        }
        
        // 保存最后一个范围
        if let (Some(start), Some(end), Some(node)) = (current_start, current_end, current_node) {
            ranges.push((start, end, node));
        }
        
        ranges
    }
}

/// 线程安全的槽位管理器
pub type AsyncSlotManager = Arc<RwLock<SlotManager>>;

// =============================================================================
// Trait实现
// =============================================================================

#[async_trait]
impl SlotManagerTrait for SlotManager {
    async fn assign_slot(&mut self, slot: u16, node_id: String) -> ClusterResult<()> {
        self.assign_slot(slot, node_id)
            .map_err(ClusterError::Config)
    }
    
    async fn unassign_slot(&mut self, slot: u16) -> ClusterResult<Option<String>> {
        self.unassign_slot(slot)
            .map_err(ClusterError::Config)
    }
    
    async fn get_slot_node(&self, slot: u16) -> Option<&String> {
        self.get_node_for_slot(slot)
    }
    
    async fn get_node_slots(&self, node_id: &str) -> Vec<u16> {
        // 返回直接的结果，而不是Future
        self.slot_to_node
            .iter()
            .enumerate()
            .filter_map(|(slot, opt_node)| {
                if let Some(node) = opt_node {
                    if node == node_id {
                        Some(slot as u16)
                    } else {
                        None
                    }
                } else {
                    None
                }
            })
            .collect()
    }
    
    async fn is_fully_covered(&self) -> bool {
        self.all_slots_assigned()
    }
    
    async fn get_unassigned_slots(&self) -> Vec<u16> {
        self.get_unassigned_slots()
    }
    
    fn calculate_slot(&self, key: &str) -> u16 {
        Self::calculate_slot(key)
    }
    
    async fn get_slot_stats(&self) -> SlotStats {
        let assigned_slots = self.assigned_slot_count() as u16;
        let unassigned_slots = CLUSTER_SLOTS - assigned_slots;
        let nodes_with_slots = self.get_slot_distribution().len();
        
        SlotStats {
            total_slots: CLUSTER_SLOTS,
            assigned_slots,
            unassigned_slots,
            nodes_with_slots,
        }
    }
    
    async fn assign_slots_batch(&mut self, slots: Vec<u16>, node_id: String) -> ClusterResult<()> {
        for slot in slots {
            self.assign_slot(slot, node_id.clone())
                .map_err(ClusterError::Config)?;
        }
        Ok(())
    }
    
    async fn rebalance_slots(&mut self, nodes: &[String]) -> ClusterResult<()> {
        self.auto_assign_slots(nodes)
            .map_err(ClusterError::Config)
    }
}

